package com.shujia.flink.asyncio

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.util.Properties

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object DemoSyncJoinMysql {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(2)

    val properties: Properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "asdasdasda")

    //创建kafkasource
    val kafkaConsumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String](
      "async",
      new SimpleStringSchema(),
      properties)

    kafkaConsumer.setStartFromLatest()

    val scoreDS: DataStream[String] = env.addSource(kafkaConsumer)

    /**
      * 关联mysql中的学生表,获取学生的信息
      *
      */

    val joinDS: DataStream[String] = scoreDS
      .keyBy(_.split(",")(0))
      .flatMap(new JoinMysqlMap)


    joinDS.print()


    env.execute()


  }
}

class JoinMysqlMap extends RichFlatMapFunction[String, String] {

  var conn: Connection = _

  override def open(parameters: Configuration): Unit = {

    Class.forName("com.mysql.jdbc.Driver")

    conn = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")

  }

  override def close(): Unit = {
    conn.close()
  }


  /**
    * flatMap  方法每一条数据执行一次 ,按顺序同步执行
    */

  override def flatMap(value: String, out: Collector[String]): Unit = {

    val start: Long = System.currentTimeMillis()
    val id: String = value.split(",")(0)

    /**
      * 通过学号查询学生信息
      *
      */





    val stat: PreparedStatement = conn.prepareStatement("select * from student where id=?")

    stat.setString(1, id)

    val set: ResultSet = stat.executeQuery()

    //模拟连接mysql花费时间
    Thread.sleep(200)

    if (set.next()) {
      val name: String = set.getString("name")

      val line: String = value + "," + name

      out.collect(line)
    }

    val end: Long = System.currentTimeMillis()

    println("关联mysql花费:" + (end - start))
  }
}